package com.shujia.flink.window;

import com.shujia.flink.event.MyEvent;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.assigners.ProcessingTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

public class Demo02Session {
    public static void main(String[] args) throws Exception {
        // 会话窗口：当一段时间没有数据，那么就认定此次会话结束并触发窗口的执行

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        DataStream<MyEvent> myDS = env.socketTextStream("master", 8888)
                .map(new MapFunction<String, MyEvent>() {
                    @Override
                    public MyEvent map(String value) throws Exception {
                        String[] split = value.split(",");
                        return new MyEvent(split[0], Long.parseLong(split[1]));
                    }
                });


        // 基于处理时间的会话窗口
        SingleOutputStreamOperator<Tuple2<String, Integer>> processSessionDS = myDS.map(e -> Tuple2.of(e.getWord(), 1), Types.TUPLE(Types.STRING, Types.INT))
                .keyBy(t2 -> t2.f0)
                .window(ProcessingTimeSessionWindows.withGap(Time.seconds(10)))
                .sum(1);

        // 基于事件时间的会话窗口
        // 指定水位线策略并提供数据中的时间戳解析规则
        SingleOutputStreamOperator<MyEvent> assDS = myDS.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<MyEvent>forMonotonousTimestamps()
                        .withTimestampAssigner((e, ts) -> e.getTs())
        );

        SingleOutputStreamOperator<Tuple2<String, Integer>> eventSessionDS = assDS.map(e -> Tuple2.of(e.getWord(), 1), Types.TUPLE(Types.STRING, Types.INT))
                .keyBy(t2 -> t2.f0)
                .window(EventTimeSessionWindows.withGap(Time.seconds(10)))
                .sum(1);

//        processSessionDS.print();
        eventSessionDS.print();
        env.execute();
    }
}
